package initialize

import (
	"fmt"
	"github.com/apache/pulsar-client-go/pulsar"
	"github.com/apache/pulsar-client-go/pulsar/log"
	"go_19/core/logrus"
	"go_19/utils"
	"time"
)

type ClientOption struct {
	URL                     string        `json:"pulsar_broker"`
	ConnectionTimeout       time.Duration `json:"connection_timeout"`
	OperationTimeout        time.Duration `json:"operation_timeout"`
	MaxConnectionsPerBroker int           `json:"max_connections_per_broker"`
	LogPath                 string        `json:"log_path"`
	LogName                 string        `json:"log_name"`
}

type TopicConsumer struct {
	ClientOption                ClientOption `json:"client_option"'`
	Topic                       string       `json:"topic"`
	SubscriptionType            int          `json:"subscription_type"`             //0:Exclusive 1:Shared 2:Failover 3:KeyShared
	SubscriptionInitialPosition int          `json:"subscription_initial_position"` //default - 0:latest 1:earliest
	SubscriptionName            string       `json:"subscription_name"`
	ConsumerName                string       `json:"consumer_name"`
}
type TopicProducer struct {
	ClientOption ClientOption `json:"client_option"'`
	Topic        string       `json:"pulsar_topic"`
	DeliverAfter int          `json:"deliver_after"`
	ProducerName string       `json:"producer_name"`
}
type TopicReader struct {
	ClientOption           ClientOption `json:"client_option"`
	Topic                  string       `json:"topic"`
	ReaderName             string       `json:"reader_name"`
	SubscriptionRolePrefix string       `json:"subscription_role_prefix"`
	SubscriptionName       string       `json:"subscription_name"`
	StartMessageID         string       `json:"start_message_id"`

	MapMessageIds string `json:"map_message_ids"`
}

func PulsarClient(clientOption ClientOption) pulsar.Client {
	clientOptions := pulsar.ClientOptions{
		URL:                     clientOption.URL,
		ConnectionTimeout:       clientOption.ConnectionTimeout,
		OperationTimeout:        clientOption.OperationTimeout,
		MaxConnectionsPerBroker: clientOption.MaxConnectionsPerBroker,
	}
	if clientOption.LogPath != "" {
		logPathExists, _ := utils.PathExists(clientOption.LogPath)
		if logPathExists {
			logName := "pulsar_" + clientOption.LogName
			//fmt.Println(logName)
			logger := logrus.CreateSimpleLogger(logName)
			clientOptions.Logger = log.NewLoggerWithLogrus(logger)
		}
	}
	client, err := pulsar.NewClient(clientOptions)
	if err != nil {
		panic(fmt.Errorf("pulsar client %s", err))
	}
	defer client.Close()
	return client
}

/*type messageID struct {
	ledgerID     int64
	entryID      int64
	batchIdx     int32
	partitionIdx int32
}

func NewMessageID(ledgerID int64, entryID int64, batchIdx int32, partitionIdx int32) pulsar.MessageID {
	return messageID{
		ledgerID:     ledgerID,
		entryID:      entryID,
		batchIdx:     batchIdx,
		partitionIdx: partitionIdx,
	}
}*/
